package top.wangjiu.product.service;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import top.wangjiu.service.interfaces.OneService;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;

public class OneServiceImpl implements OneService {

    private static final Log log = LogFactory.getLog(OneServiceImpl.class);
    @Override
    public String sayHello() {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "hello world";
    }

    @Override
    public CompletableFuture<Integer> getSum(Integer one, Integer two) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return one + two;
        });
    }

    @Override
    public Flux<Integer> getList(Integer start, Integer end) {
        AtomicInteger num = new AtomicInteger(start);
        return Flux.create(fluxSink -> {
            if (fluxSink.isCancelled()) {
                fluxSink.complete();
                return;
            }
            fluxSink.onRequest(r -> {
                System.out.println("product"+Thread.currentThread().getId()+" receive request count:"+r);
                int i = 0;
                while (i++ < r && num.getAndIncrement() < end) {
                    fluxSink.next(num.get());
                    System.out.println("product"+Thread.currentThread().getId() +":"+ num.get());
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                if(num.get()>=end){
                    //停止生产
                    fluxSink.complete();
                }
            });
        });
    }

    @Override
    public Mono<String> sayByeBye(String name) {
        return Mono.create(sink -> {
            sink.onRequest(l->{
                if(name.equals("张三")){
                    sink.success("byebye "+name);
                }else {
                    sink.success();
                }
            });
        });
    }
}

